package org.fujene.communicate.indexer;

import java.util.Map.Entry;
import java.util.TreeMap;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.LinkedBlockingQueue;

import org.fujene.communicate.hash.VirtualContentHash;
import org.fujene.communicate.lock.IndexableSwitch;
import org.fujene.communicate.node.Node;
import org.fujene.communicate.repli.ReplicateCommunicator;
import org.fujene.logging.LogParser;
import org.fujene.structure.TermIndexee;

public class TermIndexerHost implements Runnable{
	private String newTerm;
	private static int replication;
	private static Node myNode;
	private static CopyOnWriteArrayList<Node> nodes;
	private static LinkedBlockingQueue<TermIndexee> hostQueue;
	private static ConcurrentHashMap<Node, TermIndexee> suspendQueue;
	private int retryCount = 2;

	static{
		hostQueue = new LinkedBlockingQueue<TermIndexee>();
		suspendQueue = new ConcurrentHashMap<Node, TermIndexee>();
	}

	public TermIndexerHost(){}

	@Deprecated
	public TermIndexerHost(TermIndexee indexer){
		this.retryCount = 2;
	}

	public static void property(int repl, Node ownNode, CopyOnWriteArrayList<Node> list){
		replication = repl;
		myNode = ownNode;
		nodes = list;
	}

	public static void addToQueue(TermIndexee newIndex){
		hostQueue.add(newIndex);
	}

	public static boolean isEmpty(){
		return hostQueue.isEmpty();
	}

	public static int residue(){
		return hostQueue.size();
	}
	
	public static void resumeQueue(Node liveNode) {
		for(Entry<Node, TermIndexee> entries: suspendQueue.entrySet()){
			if(entries.getKey() == liveNode) {
				hostQueue.add(entries.getValue());
				suspendQueue.remove(entries);
			}
		}
	}
	
	// Drop from suspend queue
	public static void nodeDead(Node deadNode) {
		for(Entry<Node, TermIndexee> entries: suspendQueue.entrySet()){
			if(entries.getKey() == deadNode) suspendQueue.remove(entries);
		}
	}

	@Override
	public void run(){
		long time;
		TermIndexee newIndex = null;

		while(true){
			try{
				newIndex = hostQueue.take();
			}catch(InterruptedException e1){
				continue;
			}
			long hash = VirtualContentHash.getTermHash(newIndex.getTerm());
			TreeMap<Long, Node> nearestHashes = new TreeMap<Long, Node>();
			Entry<Long, Node> currentEntry = null;
			for(int i = 0; i < replication; i++)
				for(Node eachNode: nodes)
					nearestHashes.put(eachNode.getAttributeHash(hash), eachNode);

			try{
				IndexableSwitch.waitIfLocked();
			}catch(InterruptedException e){}
			for(int i = 0; i < (replication > nodes.size() ? nodes.size() : replication); i++){
				if(currentEntry == null) currentEntry = nearestHashes.ceilingEntry(Long.valueOf(hash));
				else currentEntry = nearestHashes.higherEntry(currentEntry.getKey());
				if(currentEntry == null) currentEntry = nearestHashes.firstEntry();

				if(currentEntry.getValue().equals(myNode)){
					// local: Directly adds to client queue
					TermIndexerClient.addToQueue(newIndex);
				}else if(currentEntry.getValue().isSuspend()){
					suspendQueue.put(currentEntry.getValue(), newIndex);
				}else{
					// remote
					for(int j = 0; j < retryCount; j++){
						time = LogParser.initTime();
						try{
							ReplicateCommunicator.indexTermRemote(newIndex.getTerm(), newIndex.getTermFreq(),
									newIndex.getID(), newIndex.getValueType(), currentEntry.getValue());
							break;
						}catch(Exception e){
							LogParser.warning("Can not index term \"" + newTerm
									+ "\", probably due to server down.");
							LogParser.warning("Retry to index in next node.");
						}
						LogParser.timerFinest("Index / store term remotely", time);
					}
				}
			}
		}
	}
}
